package com.stars.easyms.mq.core.rocketmq;

import com.stars.easyms.mq.core.MQMessageSender;
import com.stars.easyms.mq.dto.MQSendResult;
import com.stars.easyms.mq.exception.EasyMsMQException;
import com.stars.easyms.mq.properties.EasyMsMQProperties;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

/**
 * RocketMQ消息发送者
 *
 * @author guoguifang
 * @date 2018-04-23 13:54
 * @since 1.0.0
 */
public class RocketMQMessageSender implements MQMessageSender {

    private static final Logger logger = LoggerFactory.getLogger(RocketMQMessageSender.class);

    private DefaultMQProducer mqProducer;

    private String nameServer;

    private String groupName;

    @Override
    public MQSendResult convertAndSend(String currentSendKey, String messageStr, String destinationType, Boolean currMessIsAllowNonPersistent) {
        Message message = new Message(currentSendKey, messageStr.getBytes(StandardCharsets.UTF_8));
        try {
            SendResult result = mqProducer.send(message);
            return MQSendResult.from(result.getSendStatus() == SendStatus.SEND_OK, SendResult.encoderSendResultToJson(result));
        } catch (Exception e) {
            throw new EasyMsMQException("Message {} send fail!", message, e);
        }
    }

    private void createMQProducer() {
        try {
            mqProducer = new DefaultMQProducer(groupName);
            mqProducer.setNamesrvAddr(nameServer);
            mqProducer.setInstanceName(UUID.randomUUID().toString());
            mqProducer.setVipChannelEnabled(false);
            mqProducer.start();
        } catch (Exception e) {
            logger.error("注册MQ发送事务资源失败!", e);
        }
    }

    public RocketMQMessageSender(String applicationId, EasyMsMQProperties easyMsMQProperties) {
        this.nameServer = easyMsMQProperties.getRocketmq().getNameServer();
        this.groupName = applicationId + "_PRODUCER";
        this.createMQProducer();
    }
}
